package com.hrvsr.scheduler.distributed;

import com.hrvsr.common.redis.JedisHolder;
import com.hrvsr.common.util.PropertyUtil;
import com.hrvsr.scheduler.distributed.redisbloomfilter.RedisBloomFilter;

import redis.clients.jedis.JedisPool;

public class RedisSchedulerBuilder {
	private JedisPool jedisPool;

	// request queue config
	private int requestQueueDbNo = Integer.valueOf(PropertyUtil.getProperty("scheduler.redis.dbNo"));
	private String requestQueueName = PropertyUtil.getProperty("scheduler.redis.queue");

	// redis bloom filter conig
	private int bloomFilterDbNo = 0;
	private String bloomFilterKeyPrefix = "bloomFilter_";
	private long expectedMaxQueueSize = 10000000L;
	private double falsePositiveRate = 0.000001;

	public RedisSchedulerBuilder setExpectedMaxQueueSize(long expectedMaxQueueSize) {
		this.expectedMaxQueueSize = expectedMaxQueueSize;
		return this;
	}

	public RedisSchedulerBuilder setFalsePositiveRate(double falsePositiveRate) {
		this.falsePositiveRate = falsePositiveRate;
		return this;
	}

	public RedisSchedulerBuilder setJedisPool(JedisPool jedisPool) {
		this.jedisPool = jedisPool;
		return this;
	}

	public RedisSchedulerBuilder setRequestQueueDbNo(int requestQueueDbNo) {
		this.requestQueueDbNo = requestQueueDbNo;
		return this;
	}

	public RedisSchedulerBuilder setRequestQueueName(String requestQueueName) {
		this.requestQueueName = requestQueueName;
		return this;
	}

	public RedisSchedulerBuilder setBloomFilterDbNo(int bloomFilterDbNo) {
		this.bloomFilterDbNo = bloomFilterDbNo;
		return this;
	}

	public RedisSchedulerBuilder setBloomFilterKeyPrefix(String bloomFilterKeyPrefix) {
		this.bloomFilterKeyPrefix = bloomFilterKeyPrefix;
		return this;
	}

	public RedisScheduler build() {
		if (jedisPool == null) {
			jedisPool = JedisHolder.jedisPool();
		}
		RedisBloomFilter bloomFilter = new RedisBloomFilter(jedisPool, bloomFilterDbNo, bloomFilterKeyPrefix,
				expectedMaxQueueSize, falsePositiveRate);
		RedisBloomDuplicateRemover duplicateRemover = new RedisBloomDuplicateRemover(bloomFilter);
		RedisScheduler scheduler = new RedisScheduler();
		scheduler.setJedisPool(jedisPool);
		scheduler.setDbNo(requestQueueDbNo);
		scheduler.setQueue(requestQueueName);
		scheduler.setDuplicateRemover(duplicateRemover);
		return scheduler;
	}

}
